//package com.yunhe.common.websocket;
//
//import com.fasterxml.jackson.databind.ObjectMapper;
//import com.yunhe.common.constant.AbnormalConsts;
//import com.yunhe.common.constant.KafkaConsts;
//import com.yunhe.common.domain.data.AlarmEvent;
//import com.yunhe.common.model.PageParam;
//import com.yunhe.common.model.base.AbnormalStatusDTO;
//import com.yunhe.common.service.base.AbnormalStatusService;
//import com.yunhe.common.service.data.AlarmEventService;
//import com.yunhe.common.util.StringUtil;
//import org.apache.kafka.clients.consumer.ConsumerRecord;
//import org.slf4j.Logger;
//import org.slf4j.LoggerFactory;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.data.domain.Page;
//import org.springframework.kafka.annotation.KafkaListener;
//import org.springframework.transaction.interceptor.TransactionAspectSupport;
//import org.springframework.web.socket.CloseStatus;
//import org.springframework.web.socket.TextMessage;
//import org.springframework.web.socket.WebSocketSession;
//
//import javax.annotation.PostConstruct;
//import java.util.*;
//
///**
// * 实时异常WebSocket处理类
// *
// * @author liuronglei
// */
//public class AbnormalRealtimeHandler extends WebsocketRealtimeHandler {
//    private static final Logger logger = LoggerFactory.getLogger(AbnormalRealtimeHandler.class);
//    private static final ObjectMapper mapper = new ObjectMapper();
//    @Autowired
//    private AlarmEventService alarmEventService;
//    @Autowired
//    private AbnormalStatusService abnormalStatusService;
//    private PageParam pageParam;
//    private List<Long> abnormalStatusIds;
//
//    @PostConstruct
//    void init() {
//        // 初始化分页参数，为了避免数据量过大，只取前100条记录
//        pageParam = new PageParam();
//        pageParam.setPage(1);
//        pageParam.setSize(100);
//        // 初始化异常状态是待处理和已派单的ID列表
//        abnormalStatusIds = new ArrayList<>();
//        AbnormalStatusDTO statusReady = abnormalStatusService.findByName(AbnormalConsts.ABNORMAL_STATUS.READY.getName());
//        if (statusReady != null) {
//            abnormalStatusIds.add(statusReady.getId());
//        }
//        AbnormalStatusDTO sentReady = abnormalStatusService.findByName(AbnormalConsts.ABNORMAL_STATUS.SENT.getName());
//        if (sentReady != null) {
//            abnormalStatusIds.add(sentReady.getId());
//        }
//    }
//
//    @Override
//    void afterEstablished(WebSocketSession session) {
//
//    }
//
//    @Override
//    void afterClosed(WebSocketSession session, CloseStatus status) {
//
//    }
//
//    @Override
//    void handleMessage(WebSocketSession session, TextMessage message, List<String> points) {
//
//    }
//
//    @Override
//    Object getLastItem(String point) {
//        Long stationId = null;
//        Long deviceId = null;
//        String[] pointArr = point.split(";");
//        if (pointArr.length == 1) {
//            stationId = Long.parseLong(pointArr[0]);
//        } else if (pointArr.length == 2) {
//            stationId = Long.parseLong(pointArr[0]);
//            deviceId = Long.parseLong(pointArr[1]);
//        }
//        return ((Page<AlarmEvent>) alarmEventService.findByCondition(Optional.empty(),
//                Optional.of(Collections.singletonList(stationId)), Optional.ofNullable(deviceId),
//                Optional.empty(), Optional.empty(), Optional.of(abnormalStatusIds), Optional.empty(),
//                Optional.empty(), Optional.empty(), pageParam)).getContent();
//    }
//
//    @Override
//    Map<String, Object> getLastItemBatch(String[] pointArr) {
//        return null;
//    }
//
//    @Override
//    boolean needSend(String point, Object previousItem, Object newItem) {
//        if (newItem == null) {
//            return previousItem != null;
//        } else {
//            if (previousItem == null) {
//                return true;
//            } else {
//                List<AlarmEvent> previousList = (List<AlarmEvent>) previousItem;
//                List<AlarmEvent> newList = (List<AlarmEvent>) newItem;
//                // 按告警ID和告警状态ID来做匹配，如果都能匹配上，则认为是没有变化的
//                HashMap<Long, Long> newListMap = new HashMap<>();
//                for (AlarmEvent event : newList) {
//                    newListMap.put(event.getId(), event.getAbnormalStatusId());
//                }
//                HashMap<Long, Long> previousListMap = new HashMap<>();
//                for (AlarmEvent event : previousList) {
//                    previousListMap.put(event.getId(), event.getAbnormalStatusId());
//                }
//                return !newListMap.equals(previousListMap);
//            }
//        }
//    }
//
//    @Override
//    public String getPointField() {
//        return "key";
//    }
//
//    /**
//     * 监听告警内容刷新消息
//     * @param record 消息内容：主要是电站ID
//     */
//    @KafkaListener(topics = KafkaConsts.Topic.AlarmEventRefresh)
//    public void sendMessage(ConsumerRecord<?, String> record) {
//        try {
//            String value = record.value();
//            if (!StringUtil.isEmpty(value)) {
//                Map<String, Object> map = (Map<String, Object>) mapper.readValue(value, Map.class);
//                Object stationId = map.get("stationId");
//                if (stationId != null) {
//                    this.sendLastMessage(stationId.toString());
//                }
//            }
//        } catch (Exception e) {
//            logger.error("刷新异常按等级统计报错，{}，{}", record.value(), e);
//            TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
//        }
//    }
//}